package com.atguigu.flink.exec2;

import com.atguigu.flink.function.MyUtil;
import com.atguigu.flink.pojo.LoginEvent;
import com.atguigu.flink.pojo.UserBehavior;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.GlobalWindow;
import org.apache.flink.util.Collector;

import java.util.List;

/**
 * Created by Smexy on 2022/12/17
 *
 * 如果同一用户（可以是不同IP）在2秒之内连续两次登录，都失败，就认为存在恶意登录的风险，输出相关的信息进行报警提示
 *
 *
 * 同一用户（可以是不同IP）在2秒之内连续两次登录，都失败
 *
 * 5402,83.149.11.115,success,1558430815
 *
 *      和数量相关:
 *              2秒之内  :  2秒的窗口 不合适
 *              连续两次登录:  基于个数的窗口。 size = 2 ,slide = 1
 *                      保证数据是有序
 */
public class Demo5_Login
{
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //env.setParallelism(1);

        env
            .readTextFile("data/LoginLog.csv")
            .map(new MapFunction<String, LoginEvent>()
            {
                @Override
                public LoginEvent map(String value) throws Exception {
                    String[] words = value.split(",");
                    LoginEvent event = new LoginEvent(
                        Long.valueOf(words[0]),
                        words[1],
                        words[2],
                        Long.valueOf(words[3])
                    );
                    return event;
                }
            })
            .keyBy(LoginEvent::getUserId)
            .countWindow(2,1)
            .process(new ProcessWindowFunction<LoginEvent, String, Long, GlobalWindow>()
            {
                //elements 可能是1，也能是2
                @Override
                public void process(Long key, Context context, Iterable<LoginEvent> elements, Collector<String> out) throws Exception {

                    List<LoginEvent> loginEvents = MyUtil.parseIterable(elements);

                    //在2秒之内连续两次登录，都失败
                    if (loginEvents.size() == 2){

                        LoginEvent e1 = loginEvents.get(0);
                        LoginEvent e2 = loginEvents.get(1);

                        //对比
                        if ("fail".equals(e1.getEventType())
                              &&
                            "fail".equals(e2.getEventType())
                             &&
                            Math.abs(e1.getEventTime() - e2.getEventTime()) < 2){
                            out.collect(key + "恶意登录...." + e1.getEventTime() + ","+e2.getEventTime());
                        }

                    }
                }
            })
            .print().setParallelism(1);

        env.execute();

    }
}
